package com.teamsun.mqtt.handle;

import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttPubAckMessage;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttQoS;

import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;


import org.apache.ibatis.session.SqlSession;

import com.teamsun.mqtt.db.DBSessionFactory;
import com.teamsun.mqtt.entity.MsgRep;
import com.teamsun.mqtt.entity.SendMsg;
import com.teamsun.mqtt.service.ChannelDataService;
import com.teamsun.mqtt.service.MessageDataService;
import com.teamsun.mqtt.service.ReSendService;

/**
 * 发布消息处理类
 * @author acer
 *
 */
@Sharable
public class PushServiceHandle extends ChannelInboundHandlerAdapter {

	/**
	 * 保存消息的 mybatis 命名空间
	 */
	static final  String saveMsg="com.teamsun.mqtt.entity.MsgRepMapper.insert";
	
	/**
	 * 保存发过的消息的 mybatis 命名空间
	 */
	final   String  saveSendMsg="com.teamsun.mqtt.entity.SendMsgMapper.insert";
	
	/**
	 * 更改发过的消息的状态的 mybatis 命名空间
	 */
	final   String  updateSendMsg="com.teamsun.mqtt.entity.SendMsgMapper.updateByIdenAndMsg";
	
	/**
	 * 查询一条消息的命名空间
	 */
	final  String queryMsg="com.teamsun.mqtt.entity.MsgRepMapper.selectByExample";
	
	
	

	
	/**
	 * 用于根据登录的客户端标识找channel
	 */
	ConcurrentHashMap<String, Channel> str2channel;
	
	/**
	 * 用于根channel 找登录的客户端
	 */
	ConcurrentHashMap<Channel, String> channel2str;
	
	
	/**
	 * 消息 队列
	 */
	BlockingQueue<MsgRep> msgRepsque;
	
	/**
	 * 已经发送过的消息 队列
	 */
	
	BlockingQueue<SendMsg> msgSendedQue;

	/**
	 * 数据库操作线程组
	 */
	ExecutorService dboptgroup;
	
	AtomicInteger atomicInteger=new AtomicInteger();
	
	
	public PushServiceHandle(ExecutorService dboptgroup){
		
		
		msgRepsque=MessageDataService.getMsgRepsque();
		msgSendedQue=MessageDataService.getMsgSendedQue();
		
		str2channel=ChannelDataService.getStr2channel();
		channel2str=ChannelDataService.getChannel2str();
		
		this.dboptgroup=dboptgroup;
	  
		
	}
	
	/**
	 * 业务开始
	 */
	public void channelRead(ChannelHandlerContext ctx, Object msg) {

		if (msg instanceof MqttMessage) {

			MqttMessage message = (MqttMessage) msg;
			MqttMessageType messageType = message.fixedHeader().messageType();

			switch (messageType) {
			case PUBLISH://客户端发布普通消息
				MqttPublishMessage messagepub = (MqttPublishMessage) msg;
				pub(ctx, messagepub);
				break;
				
			case  PUBREL: //客户端发布释放
				pubrel(ctx, message);
				break;
			case PUBREC://客户端发布收到
				pubrec(ctx, message);
			default:
				ctx.fireChannelRead(msg);
				break;
			}

		}

		else
			ctx.channel().close();
	}
	
	/**
	 * 处理客户端的发布请求
	 * 根据客户端请求的QOS级别 发送相应的响应
	 * 把相应的消息存储在数据库
	 * 现在的 客户端标识是 发送方的
	 * @param ctx
	 * @param messagepub
	 */
	private void pub(final ChannelHandlerContext ctx,
			MqttPublishMessage messagepub) {
		
		 int  messageid=messagepub.variableHeader().messageId();
		String iden=channel2str.get(ctx.channel());
		
		MqttQoS mqttQoS=messagepub.fixedHeader().qosLevel();
		
		MqttFixedHeader fixedHeader=null;
		if(mqttQoS.value()<=1){
			//不是级别最高的QOS  返回 puback 即可
			fixedHeader	= new MqttFixedHeader(
					MqttMessageType.PUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0);
		}
		else{
			//否则发送发布收到  QOS级别会2
			fixedHeader	= new MqttFixedHeader(
					MqttMessageType.PUBREC, false, MqttQoS.EXACTLY_ONCE, false, 0);
		}
		
		MqttMessageIdVariableHeader connectVariableHeader = MqttMessageIdVariableHeader
				.from(messageid);

		MqttPubAckMessage ackMessage = new MqttPubAckMessage(fixedHeader,
				connectVariableHeader);
		ctx.writeAndFlush(ackMessage);
		
		
		
		final	String topname = messagepub.variableHeader().topicName();

		ByteBuf buf = messagepub.content();
		final byte[] bs = new byte[buf.readableBytes()];		
		buf.readBytes(bs);
		MsgRep msgRep=new MsgRep(messageid,topname,iden,bs);
		
		//如果不能放进去了 就清除并且提交到数据库
		 if(!msgRepsque.offer(msgRep)){
			List<MsgRep> msgReps=new ArrayList<MsgRep>();
			 msgRepsque.drainTo(msgReps);
			 msgReps.add(msgRep);
			subSaveMsg(msgReps);
		 }
		
		
	//如果只是普通的发布 QOS级别低的话就直接发送了消息
		   MessageDataService.sendPubMsg( msgRep, mqttQoS);
		try {
			System.out.println("第"+atomicInteger.incrementAndGet()+"\t条发布的内容为" +topname+ new String(bs, "UTF-8"));
			
		} catch (UnsupportedEncodingException e) {
			e.printStackTrace();
		}

		

	}
	
	
	/**
	 * 处理客户端过来的发布释放
	 * 在消息池里根据客户端标识和消息id面拿到一个消息  并且把这个消息发送出去
	 * 现在的 客户端标识是 发送方的
	 */
	private void pubrel(final ChannelHandlerContext ctx,
			MqttMessage messagepub) {
		
		MqttMessageIdVariableHeader variableHeader=(MqttMessageIdVariableHeader)messagepub.variableHeader();

		Integer messageid=variableHeader.messageId();

		MqttQoS mqttQoS=messagepub.fixedHeader().qosLevel();
		
		MqttFixedHeader fixedHeader=null;
		
		if(mqttQoS.value()<=1)
			//发送发布收到  QOS级别会1
			fixedHeader	= new MqttFixedHeader(
					MqttMessageType.PUBCOMP, false, MqttQoS.AT_MOST_ONCE, false, 0);
		else
			//否则发送发布收到  QOS级别会2
			fixedHeader	= new MqttFixedHeader(
					MqttMessageType.PUBCOMP, false, MqttQoS.EXACTLY_ONCE, false, 0);
		
		

		MqttMessageIdVariableHeader connectVariableHeader = MqttMessageIdVariableHeader
				.from(messageid);

		MqttPubAckMessage ackMessage = new MqttPubAckMessage(fixedHeader,
				connectVariableHeader);
		ctx.writeAndFlush(ackMessage);
		

	}
	
	
	/**
	 *   处理客户端 发布收到
	 *   
	 *   对客户端发送发布释放
	 *   根据 客户端收到的messageid 找到相应的message  并且  存储到消息记录里面
	 *   现在的 客户端标识是 接受方的
	 *   现在的messageid是数据库里面的主键id
	 *   最后移除重发队列 防止消息重发
	 * @param ctx
	 * @param messagepub
	 */
	private void pubrec(final ChannelHandlerContext ctx,
			MqttMessage messagepub) {
		
		MqttMessageIdVariableHeader variableHeader=(MqttMessageIdVariableHeader)messagepub.variableHeader();

		Integer messageid=variableHeader.messageId();
		
		
		MqttQoS mqttQoS=messagepub.fixedHeader().qosLevel();
		
		MqttFixedHeader fixedHeader=null;
		
		if(mqttQoS.value()<=1)
			//发送发布收到  QOS级别会1
			fixedHeader	= new MqttFixedHeader(
					MqttMessageType.PUBREL, false, MqttQoS.AT_MOST_ONCE, false, 0);
		else
			//否则发送发布收到  QOS级别会2
			fixedHeader	= new MqttFixedHeader(
					MqttMessageType.PUBREL, false, MqttQoS.EXACTLY_ONCE, false, 0);
		
		

		MqttMessageIdVariableHeader connectVariableHeader = MqttMessageIdVariableHeader
				.from(messageid);

		MqttPubAckMessage ackMessage = new MqttPubAckMessage(fixedHeader,
				connectVariableHeader);
		ctx.writeAndFlush(ackMessage);
		
		
	
		String iden=channel2str.get(ctx.channel());
		MessageDataService.updateSendMsg(iden,messageid);
		ReSendService.getIntanceo().removeReSend(iden, messageid);
		

	}
	/**
	 * 保持消息  将多余的消息放入数据库
	 * @param msgReps
	 */
	private  void  subSaveMsg(final Collection<MsgRep> msgReps){
		
		dboptgroup.submit(new Runnable() {
			
			@Override
			public void run() {

				SqlSession session= DBSessionFactory.getSqlSession();
				for(MsgRep msgRep:msgReps){	
					session.insert(saveMsg,msgRep);
				}
				session.commit();
			}
		});
	}
	
	
	
	@Override
	public void channelInactive(final ChannelHandlerContext ctx)
			throws Exception {
		super.channelInactive(ctx);
		ctx.close();
	}
	
}
